package com.eshipenet.shipnet.dp_decoder

import org.apache.flink.streaming.api.scala._

object Main {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val dpStream = EdgeChannelDataPacketStreamProvider.create(env)

    val decodedStream = dpStream
      .map(msg => Decoder.decode(msg))
      .filter(_.isSuccess)
      .map(_.get)

    decodedStream.print("decodedStream")

    decodedStream.addSink(EdgeChannelDataPacketSinkProvider.create(env))

    val name = this.getClass.getName
    env.execute(name)
  }

}
